package proto

import (
	"context"
	"fmt"
	"io"
	"regexp"
	"sync"
	"time"

	"gitee.com/zfd81/dbrs/cluster/msgpb"
	"gitee.com/zfd81/dbrs/config"

	"gitee.com/zfd81/dbrs/cluster"
	"gitee.com/zfd81/dbrs/engine"
	"gitee.com/zfd81/dbrs/sys"

	"github.com/go-redis/redis/v8"
	"github.com/sirupsen/logrus"

	"gitee.com/zfd81/dbrs/internal/token"

	pb "gitee.com/zfd81/dbrs/proto/dbrspb"
)

type BrokerSystem struct {
}

func (s *BrokerSystem) AddUser(ctx context.Context, request *pb.AddUserRequest) (*pb.ControlResponse, error) {
	_, err := parseToken(request.Token)
	if err != nil {
		return nil, err
	}
	userinfo := request.Userinfo
	if userinfo == nil {
		return nil, fmt.Errorf("User information cannot be empty.")
	}
	user := sys.NewUser(userinfo.Name, userinfo.Password)
	err = user.Save()
	if err != nil {
		logrus.Errorf("Add user '%s' error: %v\n", userinfo.Name, err)
		return nil, err
	}
	msg := cluster.NewMessage(msgpb.Action_ADD_USER)
	msg.Add(user.Name)
	msg.Add(user.Password)
	err = msg.Send()
	if err != nil {
		logrus.Errorf("Add user '%s' error: %v\n", userinfo.Name, err)
		return nil, err
	}
	logrus.Infof("Add user '%s' successfully.", userinfo.Name)
	return &pb.ControlResponse{}, nil
}
func (s *BrokerSystem) DeleteUser(ctx context.Context, request *pb.DeleteUserRequest) (*pb.ControlResponse, error) {
	_, err := parseToken(request.Token)
	if err != nil {
		return nil, err
	}
	userinfo := request.Userinfo
	if userinfo == nil || userinfo.Name == "" {
		return nil, fmt.Errorf("User name cannot be empty.")
	}

	err = sys.DeleteUser(userinfo.Name)
	if err != nil {
		logrus.Errorf("Delete user '%s' error: %v\n", userinfo.Name, err)
		return nil, err
	}

	msg := cluster.NewMessage(msgpb.Action_DEL_USER)
	msg.Add(userinfo.Name)
	err = msg.Send()
	if err != nil {
		logrus.Errorf("Delete user '%s' error: %v\n", userinfo.Name, err)
		return nil, err
	}

	logrus.Infof("Delete user '%s' successfully.", userinfo.Name)
	return &pb.ControlResponse{}, nil
}

func (s *BrokerSystem) ChangePassword(ctx context.Context, request *pb.ModifyUserRequest) (*pb.ControlResponse, error) {
	_, err := parseToken(request.Token)
	if err != nil {
		return nil, err
	}
	userinfo := request.Userinfo
	if userinfo == nil {
		return nil, fmt.Errorf("User information cannot be empty.")
	}
	user := &sys.User{
		Name:     userinfo.Name,
		Password: userinfo.Password,
	}
	err = user.Modify()
	if err != nil {
		logrus.Errorf("Modify user '%s' password error: %v\n", userinfo.Name, err)
		return nil, err
	}

	msg := cluster.NewMessage(msgpb.Action_PWD_USER)
	msg.Add(user.Name)
	msg.Add(user.Password)
	err = msg.Send()
	if err != nil {
		logrus.Errorf("Modify user '%s' password error: %v\n", userinfo.Name, err)
		return nil, err
	}

	logrus.Infof("Modify user '%s' password successfully.", userinfo.Name)
	return &pb.ControlResponse{}, nil
}

func (s *BrokerSystem) SetUserSignature(ctx context.Context, request *pb.ModifyUserRequest) (*pb.ControlResponse, error) {
	_, err := parseToken(request.Token)
	if err != nil {
		return nil, err
	}
	userinfo := request.Userinfo
	if userinfo == nil {
		return nil, fmt.Errorf("User information cannot be empty.")
	}
	user := &sys.User{
		Name:             userinfo.Name,
		DefaultSignature: userinfo.Sign,
	}
	err = user.Modify()
	if err != nil {
		logrus.Errorf("Set user '%s' signature error: %v\n", userinfo.Name, err)
		return nil, err
	}

	msg := cluster.NewMessage(msgpb.Action_SIGN_USER)
	msg.Add(user.Name)
	msg.Add(user.DefaultSignature)
	err = msg.Send()
	if err != nil {
		logrus.Errorf("Set user '%s' signature error: %v\n", userinfo.Name, err)
		return nil, err
	}

	logrus.Infof("Set user '%s' signature successfully.", userinfo.Name)
	return &pb.ControlResponse{}, nil
}

func (s *BrokerSystem) FindAllUsers(ctx context.Context, request *pb.FindAllUsersRequest) (*pb.FindAllUsersResponse, error) {
	_, err := parseToken(request.Token)
	if err != nil {
		return nil, err
	}
	users, err := sys.FindAllUsers()
	if err != nil {
		return nil, err
	}
	var name string
	var prefix bool
	if request.Userinfo != nil {
		name = request.Userinfo.Name
		prefix = request.Prefix
	}
	reg := fmt.Sprintf("^%s\\w*$", name)
	if name != "" && !prefix {
		reg = fmt.Sprintf("^%s$", name)
	}
	resp := &pb.FindAllUsersResponse{}
	for _, user := range users {
		matched, _ := regexp.MatchString(reg, user.Name)
		if matched {
			resp.Users = append(resp.Users, &pb.User{
				Name:       user.Name,
				Password:   user.Password,
				Sign:       user.DefaultSignature,
				CreateTime: user.CreateTime.Format("2006-01-02 15:04:05"),
				ModifyTime: user.ModifyTime.Format("2006-01-02 15:04:05"),
			})
		}
	}
	return resp, nil
}
func (s *BrokerSystem) TableInfo(ctx context.Context, request *pb.TableRequest) (*pb.TableResponse, error) {
	_, err := parseToken(request.Token)
	if err != nil {
		return nil, err
	}
	db, err := getDatabase(request.Database)
	if err != nil {
		return nil, err
	}
	switch db.Type() {
	case engine.Type_Redis:
		rdb := db.(*engine.RedisProvider)
		pk, err := rdb.GetPrimaryKey(request.Table)
		if err != nil {
			return nil, err
		}
		return &pb.TableResponse{
			PrimaryKey: pk,
		}, nil
	default:
		return nil, fmt.Errorf("Getting table information of the %s engine is not supported.", db.Type().String())
	}
}

func (s *BrokerSystem) Load(stream pb.BrokerSystem_LoadServer) error {
	var db engine.Provider
	channel := make(chan *pb.LoadRequest, 20)
	var flag error
	var cnt int64
	wg := sync.WaitGroup{}
	wg.Add(1)
	r, err := stream.Recv()
	if err != nil {
		return err
	}
	db, err = getDatabase(r.Database)
	if err != nil {
		return err
	}
	switch db.Type() {
	case engine.Type_Redis:
		rdb := db.(*engine.RedisProvider)
		client := rdb.Client()
		ctx := context.Background()
		go func() {
			defer wg.Done()
			for data := range channel {
				_, err = client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
					var values []interface{}
					for _, line := range data.Lines {
						values = make([]interface{}, len(data.Names)*2)
						for i, name := range data.Names {
							values[i*2] = name
							values[i*2+1] = line.Fields[i]
						}
						pipe.HSet(ctx, engine.PrimaryKey(data.Table, line.Fields[data.PrimaryKeyIndex]), values...)
						cnt++
					}
					return nil
				})
				if err != nil {
					flag = err
					return
				}
			}
		}()
	default:
		return fmt.Errorf("Database %s does not support load cmd.", r.Database)
	}
	channel <- r
	for {
		r, err = stream.Recv()
		if err == io.EOF {
			close(channel)
			break
		}
		if err != nil {
			return err
		}
		if flag != nil {
			return flag
		}
		channel <- r
	}
	wg.Wait()
	logrus.Infof("Load %d pieces of data", cnt)
	return stream.SendAndClose(&pb.LoadResponse{
		Cnt:       cnt,
		Timestamp: time.Now().Unix(),
	})
}

func parseToken(tokenString string) (*token.BrokerClaims, error) {
	return token.Parse(tokenString, "zfd1q2w3e")
}

func getToken(sign, user string) (string, error) {
	claims := token.BrokerClaims{
		Tenant: user,
		Sign:   sign,
		User:   user,
	}
	return token.Create(claims, engine.Token_Secret, engine.Token_Minute)
}

func (s *BrokerSystem) Sync(ctx context.Context, request *pb.SyncRequest) (*pb.SyncResponse, error) {
	_, err := parseToken(request.Token)
	if err != nil {
		return nil, err
	}
	token, err := getToken(request.User, request.Sign)
	if err != nil {
		return nil, err
	}
	channel := make(chan *pb.StreamRowsResponse, 20)
	var flag error
	var cnt int64
	primaryKeyIndex := -1
	var values []interface{}
	wg := sync.WaitGroup{}
	wg.Add(1)
	from, err := getDatabase(request.FromDb)
	if err != nil {
		return nil, err
	}
	switch from.Type() {
	case engine.Type_Databroker:
		from_client := from.(*engine.DatabrokerProvider).Client()
		req := &pb.StreamRowsRequest{
			Token: token,
			Query: request.GetSql(),
		}
		to, err := getDatabase(request.ToDb)
		if err != nil {
			return nil, err
		}
		switch to.Type() {
		case engine.Type_Redis:
			ctx := context.Background()
			rdb := to.(*engine.RedisProvider)
			to_client := rdb.Client()
			resp, err := from_client.StreamRows(ctx, req)
			if err != nil {
				return nil, err
			}
			for {
				entry, err := resp.Recv()
				if err == io.EOF {
					close(channel)
					break
				}
				if err != nil {
					return nil, err
				}
				if primaryKeyIndex < 0 {
					pk, err := rdb.GetPrimaryKey(request.ToTable)
					if err != nil {
						return nil, err
					}
					for i, field := range entry.Fields {
						if field.Name == pk {
							primaryKeyIndex = i
							break
						}
					}
					if primaryKeyIndex < 0 {
						return nil, fmt.Errorf("Primary key column %s not found", pk)
					}
					go func() {
						defer wg.Done()
						for data := range channel {
							_, err = to_client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
								for _, row := range data.Rows {
									values = make([]interface{}, len(data.Fields)*2)
									for i, field := range data.Fields {
										values[i*2] = field.Name
										values[i*2+1] = row.Columns[i]
									}
									pipe.HSet(ctx, engine.PrimaryKey(request.ToTable, row.Columns[primaryKeyIndex]), values...)
									cnt++
								}
								return nil
							})
							if err != nil {
								flag = err
								return
							}
						}
					}()
				}
				if flag != nil {
					return nil, flag
				}
				channel <- entry
			}
		default:
			return nil, fmt.Errorf("The source database does not support %s engine.", to.Type())
		}
	default:
		return nil, fmt.Errorf("The target database does not support %s engine.", from.Type())
	}
	wg.Wait()
	logrus.Infof("Sync %d pieces of data", cnt)
	return &pb.SyncResponse{
		Cnt:       cnt,
		Timestamp: time.Now().Unix(),
	}, nil
}

func (s *BrokerSystem) ListMembers(ctx context.Context, request *pb.MemberRequest) (*pb.Members, error) {
	_, err := parseToken(request.Token)
	if err != nil {
		return nil, err
	}

	members := []*pb.Member{}
	for _, n := range cluster.Members() {
		member := &pb.Member{
			Address:     n.Addr.String(),
			Port:        n.ServPort(),
			ServiceName: n.Name,

			StartUpTime: time.Unix(n.StartUpTime(), 0).Format("2006-01-02 15:04:05"),
		}
		if n.ServAddress() == cluster.LeaderMember().ServAddress() {
			member.Role = "Leader"
		} else {
			member.Role = "Follower"
		}
		members = append(members, member)
	}
	return &pb.Members{
		Members: members,
	}, nil
}

func getDatabase(name string) (engine.Provider, error) {
	for _, db := range config.GetConfig().Databases {
		if db.Name == name {
			return engine.NewProvider(db)
		}
	}
	return nil, fmt.Errorf("database %s does not exist", name)
}
